package com.cx.base.day25_Rocketmq.consumer.normal_consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @Author: songyahui
 * @Date: 2020/10/4 19:30
 * @Version 1.0
 */
@Slf4j
public class Consumer02 {
    public static void main(String[] args) {
        consumer();
    }

    public static void consumer() {
        //        1 创建消费者Consumer ，并确定消费者组名
//                  允许追踪消息：true
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group2", true);
        //        2    设置msgtTrace点
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//        3 指定NameServer地址
        consumer.setNamesrvAddr("192.168.88.133:9876;192.168.88.134:9876");
//        consumer.setNamesrvAddr("192.168.48.128:9876;192.168.48.137:9876");
//        consumer.setNamesrvAddr("106.52.131.197:9876;47.96.165.93:9876");
        log.debug("默认最小线程数：{}", String.valueOf(consumer.getConsumeThreadMin()));
        consumer.setConsumeThreadMax(1);
        consumer.setConsumeThreadMin(1);
        System.out.println(System.nanoTime());
        log.debug("修改后最大线程数：{}", String.valueOf(consumer.getConsumeThreadMax()));
        log.debug("修改后最小线程数：{}", String.valueOf(consumer.getConsumeThreadMin()));

        try {
//            4 订阅主题Topic和Tag
            consumer.subscribe("base", "tag5 || tag_async");
//            设置订阅模式，有负载均衡、广播两种模式，默认负载均衡,下面设置的是广播
//            consumer.setMessageModel(MessageModel.BROADCASTING);
//        5 设置回调函数，处理消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                //            接受消息内容
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        try {
                            Thread.sleep(10000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println(msg.toString());
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

//        6 启动消息内容
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
